package com.ruoyi.common.net.mqtt;

import java.util.Iterator;
import java.util.List;

import com.ruoyi.common.utils.uuid.UUID;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.handler.annotation.Header;
/**
 * @author xumh
 * @title: MqttConfiguration
 * @description: TODO
 * @date 2024/12/1814:58
 */
@Configuration
@IntegrationComponentScan
@ConditionalOnProperty(
        name = {"spring.mqtt.enable"},
        matchIfMissing = false,
        havingValue = "1"
)
public class MqttConfiguration {
    @Value("${spring.mqtt.username}")
    private String username;
    @Value("${spring.mqtt.password}")
    private String password;
    @Value("${spring.mqtt.url}")
    private String hostUrl;
    @Value("${spring.mqtt.client.id}")
    private String clientId;
    @Value("${spring.mqtt.default.topic}")
    private String defaultTopic;
    @Value("${spring.mqtt.completionTimeout}")
    private int completionTimeout;
    Logger logger = LoggerFactory.getLogger(MqttConfiguration.class);
    @Autowired
    private MessageSubscribe messageHandler;
    @Autowired(required = false)
    private List<IMessageHandler> messageHandlers;

    public MqttConfiguration() {
    }

    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        if (StringUtils.isNoneBlank(new CharSequence[]{this.username})) {
            mqttConnectOptions.setUserName(this.username);
        }

        if (StringUtils.isNoneBlank(new CharSequence[]{this.password})) {
            mqttConnectOptions.setPassword(this.password.toCharArray());
        }

        String[] uris = this.hostUrl.split(",");
        mqttConnectOptions.setServerURIs(uris);
        mqttConnectOptions.setKeepAliveInterval(60);
        mqttConnectOptions.setAutomaticReconnect(true);
        mqttConnectOptions.setConnectionTimeout(30);
        mqttConnectOptions.setCleanSession(true);
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(this.getMqttConnectOptions());
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(this.clientId + "_" + UUID.randomUUID(), this.mqttClientFactory(), new String[]{this.defaultTopic});
        adapter.setCompletionTimeout((long)this.completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(new int[]{1});
        adapter.setOutputChannel(this.mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(
            inputChannel = "mqttInputChannel"
    )
    public MessageHandler handler() {
        return new MessageHandler() {
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                MqttConfiguration.this.logger.debug("message of topic:{},message:{}", topic, message.getPayload().toString());
                if (MqttConfiguration.this.messageHandlers != null) {
                    Iterator var3 = MqttConfiguration.this.messageHandlers.iterator();

                    while(var3.hasNext()) {
                        IMessageHandler handler = (IMessageHandler)var3.next();
                        handler.handle(message);
                    }

                }
            }
        };
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(
            inputChannel = "mqttOutboundChannel"
    )
    public MqttPahoMessageHandler mqttOutbound(MqttPahoClientFactory clientFactory) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(this.clientId + "_outbound", clientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(this.defaultTopic);
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultRetained(false);
        messageHandler.setAsyncEvents(false);
        return messageHandler;
    }

    @MessagingGateway(
            defaultRequestChannel = "mqttOutboundChannel"
    )
    public interface MqttMessageGateway {
        void sendMessage(@Header("mqtt_topic") String topic, String msg);

        void sendMessage(@Header("mqtt_topic") String topic, byte[] payload);

        void sendMessage(@Header("mqtt_topic") String topic, Object payload);

        void sendMessage(@Header("mqtt_topic") String topic, @Header("mqtt_qos") Integer qos, String msg);
    }
}

